package com.xxd.dfs.backupnode;

import com.xxd.dfs.namenode.rpc.model.FetchEditLogRequest;
import com.xxd.dfs.namenode.rpc.model.FetchEditLogResponse;
import com.xxd.dfs.namenode.rpc.model.UpdateCheckpointTxidRequest;
import com.xxd.dfs.namenode.rpc.service.NameNodeServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.netty.NegotiationType;
import io.grpc.netty.NettyChannelBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-08-24 11:54
 * @description:
 */
public class NameNodeRpcClient {

    private final Logger logger = LoggerFactory.getLogger(NameNodeRpcClient.class);

    private Configuration configuration = Configuration.getInstance();
    private Integer NAMENODE_PORT = Integer.parseInt(configuration.get("namenode.port"));
    private String NAMENODE_HOSTNAME = configuration.get("namenode.hostname");

    private NameNodeServiceGrpc.NameNodeServiceBlockingStub namenode;

    public NameNodeRpcClient() {
        ManagedChannel channel = NettyChannelBuilder
                .forAddress(NAMENODE_HOSTNAME, NAMENODE_PORT)
                .negotiationType(NegotiationType.PLAINTEXT)
                .build();
        this.namenode = NameNodeServiceGrpc.newBlockingStub(channel);
        logger.info("与namenode hostname = {} , port = {} 建立连接", NAMENODE_HOSTNAME, NAMENODE_PORT);
    }

    /**
     * 拉取edit log
     */
    public FetchEditLogResponse fetchEditLogs(long seq) {
        FetchEditLogRequest request = FetchEditLogRequest.newBuilder()
                .setSeq(seq)
                .build();

        return namenode.fetchEditLogs(request);
    }

    /**
     *
     * @param txid
     */
    public void updateCheckpointTxid(long txid) {
        UpdateCheckpointTxidRequest request = UpdateCheckpointTxidRequest.newBuilder()
                .setTxid(txid)
                .build();

        namenode.updateCheckpointTxid(request);
    }
}